package com.bbx.flink.table_api_or_sql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * It is also possible to create a TABLE as known from relational databases from a connector declaration.
 * The connector describes the external system that stores the data of a table. Storage systems such as
 * Apache Kafka or a regular file system can be declared here.
 */
@Slf4j
public class ConnectorTableDemo {


    /**
     * @param args
     * @throws Exception
     */

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        tableEnv.useCatalog("");



        tableEnv.executeSql("CREATE TABLE flink_user (\n" +
                "  id STRING, name STRING ,age INT,gender STRING  \n" +
                ") WITH (\n" +
                "  'connector.type' = 'jdbc', -- required: specify this table type is jdbc\n" +
                "  'connector.url' = 'jdbc:mysql://192.168.10.7:3306/flink_user?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC', -- required: JDBC DB url\n" +
                "  'connector.table' = 'user',  -- required: jdbc table name\n" +
                "  'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
                "  'connector.username' = 'root',\n" +
                "  'connector.password' = '123456'" +
                ")"

        );

//        tableResult.print();

//        String  insertSql = "insert into flink_user values ('3','a',22,'a')";
//        TableResult insertTableResult1 = tableEnv.executeSql(insertSql);
//        insertTableResult1.print();

        String querySql="select id,name,age,gender from  flink_user  where id >3";
        TableResult tableResult = tableEnv.executeSql(querySql);
        tableResult.print();

        //query table  api
        Table table = tableEnv.from("flink_user");
        Table selectTable = table.filter($("id").isGreater("1"))
                .groupBy($("id"),$("name"))
                .select($("id"), $("name"), $("age").sum().as("aaaa"))
                    ;

//        tableEnv.toRetractStream(selectTable, Row.class).print("selectTableSum");

        //query SQL
        String querySQL = "select id,name,sum(age) as bbb from   flink_user where id> '1' group by id,name";
        Table sqlTable = tableEnv.sqlQuery(querySQL);
        tableEnv.toRetractStream(sqlTable,Row.class).print("sqlTable");


        // 此处需要注释，不明白
        //Since 1.11 version, sqlUpdate() method and insertInto() method are deprecated.
        // If the Table program is built from these two methods, we must use StreamTableEnvironment.execute() method
        // instead of StreamExecutionEnvironment.execute() method to execute it.
        env.execute();
    }


}
